package com.lxp.test;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

@Slf4j
public class MqttUtil {
    public static MqttClient connectMqtt(String url, String clientId, String userName, String password) throws MqttException {
        MemoryPersistence persistence = new MemoryPersistence();
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(false);
        connOpts.setUserName(userName);
        connOpts.setPassword(password.toCharArray());
        connOpts.setConnectionTimeout(10);
        connOpts.setKeepAliveInterval(20);
        connOpts.setAutomaticReconnect(true);
        MqttClient mqttClient = new MqttClient(url, clientId,persistence);
        mqttClient.setCallback(new OnMessageCallback());
        mqttClient.connect(connOpts);
        return mqttClient;
    }

    public static void sendData(MqttClient mqttClient, String topic, String msg) throws MqttException {
        sendData(mqttClient, topic, msg, 1);
    }

    public static void sendData(MqttClient mqttClient, String topic, String msg, int qos) throws MqttException {
        MqttMessage message = new MqttMessage(msg.getBytes());
        message.setQos(qos);
        message.setRetained(false);
        mqttClient.publish(topic, message);
        log.info("发送成功");
    }

    @SneakyThrows
    public static void closeMqtt(MqttClient mqttClient) {
        if (mqttClient != null) {
            mqttClient.disconnect();
        }
    }
}
